package com.hbwxz.rabbitmq;

import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hbwxz.constant.Constants;
import com.hbwxz.utils.RabbitMqUtils;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * RabbitMQ 小案例【原生 RabbitMQ，AMQP 导 jar 包的方式搞，生产者 ➕ 消费者】
 * <p>
 * 这里不知道为啥加了 Slf4j 打印日志不出来，不知道是不是不是 javaEE 项目的原因
 *
 * @author shenzw
 * @date 2023/11/9
 */
@Slf4j
public class RabbitMqFaceOne {
    public static void main(String[] args) throws IOException, TimeoutException {

        // 获取 channel
        Channel channel = RabbitMqUtils.getChannel();

        RabbitMqUtils.installRabbitMq(Constants.MY_EXCHANGE, Constants.DIRECT_EXCHANGE, Constants.MY_QUEUE, Constants.MY_ROUTING_KEY);

        // 我的生产者
        myProducer(channel);

        // 我的消费者
        myConsumer(channel);

        // 关闭连接
        RabbitMqUtils.close(channel);

    }

    /**
     * 我的消费者
     *
     * @param channel 信道
     */
    private static void myConsumer(Channel channel) throws IOException {
        // 消费消息（队列，是否自动确认消费消息【关闭，不然消息没有正常消费的情况下也会自动确认】，交换机，消费者【具体执行消费的相关逻辑】）
        channel.basicConsume(Constants.MY_QUEUE, false, Constants.MY_EXCHANGE, new DefaultConsumer(channel) {
            /**
             * 消费消息的核心方法，覆盖父类方法
             * @param consumerTag 消息者 ID
             * @param envelope Envelope（存放消息唯一 ID 等一些相关信息）
             * @param properties AMQP.BasicProperties
             * @param body 消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                long deliveryTag = envelope.getDeliveryTag(); // 唯一的 ID
                log.info("MyConsumer handleDelivery receive message = {}，deliveryTag = {}", new String(body), deliveryTag);
                System.out.println("MyConsumer handleDelivery receive message = " + new String(body) + "，deliveryTag = " + deliveryTag);
                channel.basicAck(deliveryTag, true); // 手动 Ack
            }
        });
    }

    /**
     * 我的生产者
     *
     * @param channel 信道
     */
    private static void myProducer(Channel channel) throws IOException {

        // 演示路由失败的回调
        // 如果交换机没有正确路由到队列中则会走这个方法，前提是需要开启一个东西 mandatory = true（再发布消息的时候“第三个参数”指定）
        channel.addReturnListener((replyCode, replyText, exchange, routingKey, basicProperties, bytes) -> {
            System.out.println("Message returned with reply code: " + replyCode);
            System.out.println("Reply text: " + replyText);
            System.out.println("Exchange: " + exchange);
            System.out.println("Routing key: " + routingKey);
            System.out.println("Properties: " + basicProperties);
            System.out.println("Body: " + new String(bytes));
            log.info("Message returned with reply code = {}", replyCode);
            log.info("Reply text = {}", replyText);
            log.info("Exchange = {}", exchange);
            log.info("Routing key = {}", routingKey);
            log.info("Properties = {}", basicProperties);
            log.info("Body = {}", new String(bytes));
        });

        channel.addReturnListener(aReturn -> {
            try {
                log.info("Jackson return = {}", new ObjectMapper().writeValueAsString(aReturn));
                log.info("hutool jsonUtil return = {}", JSONUtil.toJsonStr(aReturn));
                System.out.println("Jackson return = " + new ObjectMapper().writeValueAsString(aReturn));
                System.out.println("hutool jsonUtil return = " + JSONUtil.toJsonStr(aReturn));
                System.out.println("receive message = " + new String(aReturn.getBody()));
            } catch (JsonProcessingException e) {
                log.error("Jackson convert error = ", e);
            }
        });

        // 消费者手动 Ack 了，如果没有开启 channel.confirmSelect(); 不会进行回调操作
        channel.confirmSelect();
        // 可以使用这个 listener 异步的获取消费者的 Ack 消息
        channel.addConfirmListener(new ConfirmListener() {
            /**
             * 消费者 Ack 消息之后回调
             * @param l 唯一 ID
             * @param b boolean
             */
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                System.out.println("Producer addConfirmListener handleAck listener tagId = " + l);
            }

            /**
             * 消费者 Nack 消息之后回调
             * @param l 唯一 ID
             * @param b boolean
             */
            @Override
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("Producer addConfirmListener handleNack listener tagId = " + l);
            }
        });

        byte[] msg = "Hello RabbitMQ".getBytes();

        // 交换机名称，路由 KEY，mandatory， AMQP.BasicProperties()，发送消息
        // mandatory = true时，如果交换机没有根据 routingKey 找到对应的 queue，那么交换机可以把这条消息返回给生产者，添加 returnCallback
        channel.basicPublish(Constants.MY_EXCHANGE, Constants.MY_ROUTING_KEY, true,
                new AMQP.BasicProperties().builder()
                        .deliveryMode(2) // 消息持久化设置
                        .contentType("text/plain")
                        .build(),
                msg);
    }
}
